package com.shujia.dw

import java.text.SimpleDateFormat

import com.shujia.spark.SparkTool
import org.apache.spark.sql.hive.HiveContext
import java.util.Date

import com.shujia.grld.Grid
import com.shujia.bean.CaseClass.Staypoint
import org.apache.spark.sql.SaveMode

object DwStayPointApp extends SparkTool {
  /**
    * 编写业务逻辑
    *
    */
  override def run(): Unit = {
    /**
      * 计算停留表
      *
      * 将同一个人在同一个网格的多条数据合并成一条数据
      *
      */

    val hiveContext = new HiveContext(sc)

    import hiveContext.implicits._

    val staypointPath = Constants.STAYPOINT_PATH + Constants.PARTITION_NAME + day_id
    logger.info("停留表输出路径：" + staypointPath)

    val sql =
      s"""
         |select mdn,start_time,grid_id,county_id from
         |dwi.dwi_res_regn_mergelocation_msk_d
         |where day_id=$day_id
         |
      """.stripMargin


    val df = hiveContext.sql(sql)

    val kvRDD = df.map(row => {
      val mdn = row.getAs[String]("mdn")
      val start_time = row.getAs[String]("start_time")
      val grid_id = row.getAs[String]("grid_id")
      val county_id = row.getAs[String]("county_id")

      val time = start_time.split(",")(0)

      //手机号和网格id作为key   停留时间作为value
      (mdn + "-" + grid_id + "-" + county_id, time)
    })


    kvRDD.groupByKey()
      .map(kv => {
        val mdnAndGridAndcounty_id = kv._1

        val split = mdnAndGridAndcounty_id.split("-")
        val mdn = split(0)
        val grid_id = split(1)
        val county_id = split(2)

        val format = new SimpleDateFormat("yyyyMMddhhmmss")

        val times = kv._2.map(time => {
          val date: Date = format.parse(time)
          //取出时间戳
          date.getTime
        })

        val sortTime = times.toList.sortBy(time => time)

        val startTime = sortTime.head
        val endTime = sortTime.last

        val duration = Math.abs(endTime - startTime) / 60000


        val startDate = format.format(new Date(startTime))
        val endDate = format.format(new Date(endTime))


        //取出网格中心点的经纬度
        val point = Grid.getCenter(grid_id.toLong)
        val longi = point.getX
        val lati = point.getY


        Staypoint(mdn, longi.toString, lati.toString, grid_id, county_id, duration.toString, startDate, endDate)
      })
      .toDF()
      .write
      .mode(SaveMode.Overwrite)
      .parquet(staypointPath)


  }

  /**
    * 初始化 设置 spark 运行参数
    *
    */
  override def init(): Unit = {

  }

}
